Node / meshtastic / Meshtastic-Android / files / app / src / main / java / com / geeksville / mesh / repository / network / MQTTRepository.kt

Displaying Raw • Download

T8b949e/*

* Copyright (c) 2025 Meshtastic LLC

*

* This program is free software: you can redistribute it and/or modify

* it under the terms of the GNU General Public License as published by

* the Free Software Foundation, either version 3 of the License, or

* (at your option) any later version.

*

* This program is distributed in the hope that it will be useful,

* but WITHOUT ANY WARRANTY; without even the implied warranty of

* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

* GNU General Public License for more details.

*

* You should have received a copy of the GNU General Public License

* along with this program. If not, see <https://www.gnu.org/licenses/>.

*/

Tff7b72package T7ee787com.geeksville.mesh.repository.network

Tff7b72import T7ee787com.geeksville.mesh.util.ignoreException

Tff7b72import T7ee787com.google.protobuf.ByteString

Tff7b72import T7ee787kotlinx.coroutines.channels.awaitClose

Tff7b72import T7ee787kotlinx.coroutines.flow.Flow

Tff7b72import T7ee787kotlinx.coroutines.flow.callbackFlow

Tff7b72import T7ee787kotlinx.coroutines.flow.first

Tff7b72import T7ee787org.eclipse.paho.client.mqttv3.DisconnectedBufferOptions

Tff7b72import T7ee787org.eclipse.paho.client.mqttv3.IMqttDeliveryToken

Tff7b72import T7ee787org.eclipse.paho.client.mqttv3.MqttAsyncClient

Tff7b72import T7ee787org.eclipse.paho.client.mqttv3.MqttAsyncClient.generateClientId

Tff7b72import T7ee787org.eclipse.paho.client.mqttv3.MqttCallbackExtended

Tff7b72import T7ee787org.eclipse.paho.client.mqttv3.MqttConnectOptions

Tff7b72import T7ee787org.eclipse.paho.client.mqttv3.MqttMessage

Tff7b72import T7ee787org.eclipse.paho.client.mqttv3.persist.MemoryPersistence

Tff7b72import T7ee787org.meshtastic.core.data.repository.NodeRepository

Tff7b72import T7ee787org.meshtastic.core.data.repository.RadioConfigRepository

Tff7b72import T7ee787org.meshtastic.core.model.util.subscribeList

Tff7b72import T7ee787org.meshtastic.proto.MeshProtos.MqttClientProxyMessage

Tff7b72import T7ee787org.meshtastic.proto.mqttClientProxyMessage

Tff7b72import T7ee787timber.log.Timber

Tff7b72import T7ee787java.net.URI

Tff7b72import T7ee787java.security.SecureRandom

Tff7b72import T7ee787javax.inject.Inject

Tff7b72import T7ee787javax.inject.Singleton

Tff7b72import T7ee787javax.net.ssl.SSLContext

Tff7b72import T7ee787javax.net.ssl.TrustManager

Tf0883e@Singleton

Tff7b72class T56d364MQTTRepository

Tf0883e@Inject

Tff7b72constructorTb4b4b4(

Tff7b72private Tff7b72val Te6edf3radioConfigRepositoryTb4b4b4: Te6edf3RadioConfigRepositoryTb4b4b4,

Tff7b72private Tff7b72val Te6edf3nodeRepositoryTb4b4b4: Te6edf3NodeRepositoryTb4b4b4,

Tb4b4b4) Tb4b4b4{

Tff7b72companion Tff7b72object Tb4b4b4{

T8b949e/**

* Quality of Service (QoS) levels in MQTT:

* - QoS 0: "at most once". Packets are sent once without validation if it has been received.

* - QoS 1: "at least once". Packets are sent and stored until the client receives confirmation from the server.

* MQTT ensures delivery, but duplicates may occur.

* - QoS 2: "exactly once". Similar to QoS 1, but with no duplicates.

*/

Tff7b72private Tff7b72const Tff7b72val Te6edf3DEFAULT_QOS Tff7b72= T79c0ff1

Tff7b72private Tff7b72const Tff7b72val Te6edf3DEFAULT_TOPIC_ROOT Tff7b72= Ta5d6ff"Ta5d6ffmshTa5d6ff"

Tff7b72private Tff7b72const Tff7b72val Te6edf3DEFAULT_TOPIC_LEVEL Tff7b72= Ta5d6ff"Ta5d6ff/2/e/Ta5d6ff"

Tff7b72private Tff7b72const Tff7b72val Te6edf3JSON_TOPIC_LEVEL Tff7b72= Ta5d6ff"Ta5d6ff/2/json/Ta5d6ff"

Tff7b72private Tff7b72const Tff7b72val Te6edf3DEFAULT_SERVER_ADDRESS Tff7b72= Ta5d6ff"Ta5d6ffmqtt.meshtastic.orgTa5d6ff"

Tb4b4b4}

Tff7b72private Tff7b72var Te6edf3mqttClientTb4b4b4: Te6edf3MqttAsyncClient? Tff7b72= Tff7b72null

Tff7b72fun Td2a8ffdisconnectTb4b4b4(Tb4b4b4) Tb4b4b4{

Te6edf3TimberTb4b4b4.Te6edf3iTb4b4b4(Ta5d6ff"Ta5d6ffMQTT DisconnectedTa5d6ff"Tb4b4b4)

Te6edf3mqttClientTff7b72?.Te6edf3apply Tb4b4b4{

Te6edf3ignoreException Tb4b4b4{ Te6edf3disconnectTb4b4b4(Tb4b4b4) Tb4b4b4}

Te6edf3closeTb4b4b4(Tff7b72trueTb4b4b4)

Te6edf3mqttClient Tff7b72= Tff7b72null

Tb4b4b4}

Tb4b4b4}

Tff7b72val Te6edf3proxyMessageFlowTb4b4b4: Te6edf3FlowTff7b72<Te6edf3MqttClientProxyMessageTff7b72> Tff7b72= Te6edf3callbackFlow Tb4b4b4{

Tff7b72val Te6edf3ownerId Tff7b72= Ta5d6ff"Ta5d6ffMeshtasticAndroidMqttProxy-Tffd700${Te6edf3nodeRepositoryTb4b4b4.Te6edf3myIdTb4b4b4.Te6edf3value Tff7b72?: Te6edf3generateClientIdTb4b4b4(Tb4b4b4)Tffd700}Ta5d6ff"

Tff7b72val Te6edf3channelSet Tff7b72= Te6edf3radioConfigRepositoryTb4b4b4.Te6edf3channelSetFlowTb4b4b4.Te6edf3firstTb4b4b4(Tb4b4b4)

Tff7b72val Te6edf3mqttConfig Tff7b72= Te6edf3radioConfigRepositoryTb4b4b4.Te6edf3moduleConfigFlowTb4b4b4.Te6edf3firstTb4b4b4(Tb4b4b4)Tb4b4b4.Te6edf3mqtt

Tff7b72val Te6edf3sslContext Tff7b72= Te6edf3SSLContextTb4b4b4.Te6edf3getInstanceTb4b4b4(Ta5d6ff"Ta5d6ffTLSTa5d6ff"Tb4b4b4)

T8b949e// Create a custom SSLContext that trusts all certificates

Te6edf3sslContextTb4b4b4.Te6edf3initTb4b4b4(Tff7b72nullTb4b4b4, Te6edf3arrayOfTff7b72<Te6edf3TrustManagerTff7b72>Tb4b4b4(Te6edf3TrustAllX509TrustManagerTb4b4b4(Tb4b4b4)Tb4b4b4)Tb4b4b4, Te6edf3SecureRandomTb4b4b4(Tb4b4b4)Tb4b4b4)

Tff7b72val Te6edf3rootTopic Tff7b72= Te6edf3mqttConfigTb4b4b4.Te6edf3rootTb4b4b4.Te6edf3ifEmpty Tb4b4b4{ Te6edf3DEFAULT_TOPIC_ROOT Tb4b4b4}

Tff7b72val Te6edf3connectOptions Tff7b72=

Te6edf3MqttConnectOptionsTb4b4b4(Tb4b4b4)Tb4b4b4.Te6edf3apply Tb4b4b4{

Te6edf3userName Tff7b72= Te6edf3mqttConfigTb4b4b4.Te6edf3username

Te6edf3password Tff7b72= Te6edf3mqttConfigTb4b4b4.Te6edf3passwordTb4b4b4.Te6edf3toCharArrayTb4b4b4(Tb4b4b4)

Te6edf3isAutomaticReconnect Tff7b72= Tff7b72true

Tff7b72if Tb4b4b4(Te6edf3mqttConfigTb4b4b4.Te6edf3tlsEnabledTb4b4b4) Tb4b4b4{

Te6edf3socketFactory Tff7b72= Te6edf3sslContextTb4b4b4.Te6edf3socketFactory

Tb4b4b4}

Tb4b4b4}

Tff7b72val Te6edf3bufferOptions Tff7b72=

Te6edf3DisconnectedBufferOptionsTb4b4b4(Tb4b4b4)Tb4b4b4.Te6edf3apply Tb4b4b4{

Te6edf3isBufferEnabled Tff7b72= Tff7b72true

Te6edf3bufferSize Tff7b72= T79c0ff5T79c0ff1T79c0ff2

Te6edf3isPersistBuffer Tff7b72= Tff7b72false

Te6edf3isDeleteOldestMessages Tff7b72= Tff7b72true

Tb4b4b4}

Tff7b72val Te6edf3callback Tff7b72=

Tff7b72object Tb4b4b4: T56d364MqttCallbackExtended Tb4b4b4{

Tff7b72override Tff7b72fun Td2a8ffconnectCompleteTb4b4b4(Te6edf3reconnectTb4b4b4: Tffa657BooleanTb4b4b4, Te6edf3serverURITb4b4b4: Tffa657StringTb4b4b4) Tb4b4b4{

Te6edf3TimberTb4b4b4.Te6edf3iTb4b4b4(Ta5d6ff"Ta5d6ffMQTT connectComplete: Tffd700$Te6edf3serverURITa5d6ff reconnect: Tffd700$Te6edf3reconnectTa5d6ff"Tb4b4b4)

Te6edf3channelSetTb4b4b4.Te6edf3subscribeList

Tb4b4b4.Te6edf3ifEmpty Tb4b4b4{

Tff7b72return

Tb4b4b4}

Tb4b4b4.Te6edf3forEach Tb4b4b4{ Te6edf3globalId Tff7b72-Tff7b72>

Te6edf3subscribeTb4b4b4(Ta5d6ff"Tffd700$Te6edf3rootTopicTffd700$Te6edf3DEFAULT_TOPIC_LEVELTffd700$Te6edf3globalIdTa5d6ff/+Ta5d6ff"Tb4b4b4)

Tff7b72if Tb4b4b4(Te6edf3mqttConfigTb4b4b4.Te6edf3jsonEnabledTb4b4b4) Te6edf3subscribeTb4b4b4(Ta5d6ff"Tffd700$Te6edf3rootTopicTffd700$Te6edf3JSON_TOPIC_LEVELTffd700$Te6edf3globalIdTa5d6ff/+Ta5d6ff"Tb4b4b4)

Tb4b4b4}

Te6edf3subscribeTb4b4b4(Ta5d6ff"Tffd700$Te6edf3rootTopicTffd700${Te6edf3DEFAULT_TOPIC_LEVELTffd700}Ta5d6ffPKI/+Ta5d6ff"Tb4b4b4)

Tb4b4b4}

Tff7b72override Tff7b72fun Td2a8ffconnectionLostTb4b4b4(Te6edf3causeTb4b4b4: Te6edf3ThrowableTb4b4b4) Tb4b4b4{

Te6edf3TimberTb4b4b4.Te6edf3iTb4b4b4(Ta5d6ff"Ta5d6ffMQTT connectionLost cause: Tffd700$Te6edf3causeTa5d6ff"Tb4b4b4)

Tff7b72if Tb4b4b4(Te6edf3cause Tff7b72is Te6edf3IllegalArgumentExceptionTb4b4b4) Te6edf3closeTb4b4b4(Te6edf3causeTb4b4b4)

Tb4b4b4}

Tff7b72override Tff7b72fun Td2a8ffmessageArrivedTb4b4b4(Te6edf3topicTb4b4b4: Tffa657StringTb4b4b4, Te6edf3messageTb4b4b4: Te6edf3MqttMessageTb4b4b4) Tb4b4b4{

Te6edf3trySendTb4b4b4(

Te6edf3mqttClientProxyMessage Tb4b4b4{

Tff7b72thisTb4b4b4.Te6edf3topic Tff7b72= Te6edf3topic

Te6edf3data Tff7b72= Te6edf3ByteStringTb4b4b4.Te6edf3copyFromTb4b4b4(Te6edf3messageTb4b4b4.Te6edf3payloadTb4b4b4)

Te6edf3retained Tff7b72= Te6edf3messageTb4b4b4.Te6edf3isRetained

Tb4b4b4}Tb4b4b4,

Tb4b4b4)

Tb4b4b4}

Tff7b72override Tff7b72fun Td2a8ffdeliveryCompleteTb4b4b4(Te6edf3tokenTb4b4b4: Te6edf3IMqttDeliveryToken?Tb4b4b4) Tb4b4b4{

Te6edf3TimberTb4b4b4.Te6edf3iTb4b4b4(Ta5d6ff"Ta5d6ffMQTT deliveryComplete messageId: Tffd700${Te6edf3tokenTff7b72?.Te6edf3messageIdTffd700}Ta5d6ff"Tb4b4b4)

Tb4b4b4}

Tb4b4b4}

Tff7b72val Te6edf3scheme Tff7b72= Tff7b72if Tb4b4b4(Te6edf3mqttConfigTb4b4b4.Te6edf3tlsEnabledTb4b4b4) Ta5d6ff"Ta5d6ffsslTa5d6ff" Tff7b72else Ta5d6ff"Ta5d6fftcpTa5d6ff"

Tff7b72val Tb4b4b4(Te6edf3hostTb4b4b4, Te6edf3portTb4b4b4) Tff7b72=

Te6edf3mqttConfigTb4b4b4.Te6edf3address

Tb4b4b4.Te6edf3ifEmpty Tb4b4b4{ Te6edf3DEFAULT_SERVER_ADDRESS Tb4b4b4}

Tb4b4b4.Te6edf3splitTb4b4b4(Ta5d6ff"Ta5d6ff:Ta5d6ff"Tb4b4b4, Te6edf3limit Tff7b72= T79c0ff2Tb4b4b4)

Tb4b4b4.Te6edf3let Tb4b4b4{ Tffa657itTff7b72[T79c0ff0Tff7b72] Te6edf3to Tb4b4b4(Tffa657itTb4b4b4.Te6edf3getOrNullTb4b4b4(T79c0ff1Tb4b4b4)Tff7b72?.Te6edf3toIntOrNullTb4b4b4(Tb4b4b4) Tff7b72?: Tff7b72-T79c0ff1Tb4b4b4) Tb4b4b4}

Te6edf3mqttClient Tff7b72=

Te6edf3MqttAsyncClientTb4b4b4(Te6edf3URITb4b4b4(Te6edf3schemeTb4b4b4, Tff7b72nullTb4b4b4, Te6edf3hostTb4b4b4, Te6edf3portTb4b4b4, Ta5d6ff"Ta5d6ff"Tb4b4b4, Ta5d6ff"Ta5d6ff"Tb4b4b4, Ta5d6ff"Ta5d6ff"Tb4b4b4)Tb4b4b4.Te6edf3toStringTb4b4b4(Tb4b4b4)Tb4b4b4, Te6edf3ownerIdTb4b4b4, Te6edf3MemoryPersistenceTb4b4b4(Tb4b4b4)Tb4b4b4)Tb4b4b4.Te6edf3apply Tb4b4b4{

Te6edf3setCallbackTb4b4b4(Te6edf3callbackTb4b4b4)

Te6edf3setBufferOptsTb4b4b4(Te6edf3bufferOptionsTb4b4b4)

Te6edf3connectTb4b4b4(Te6edf3connectOptionsTb4b4b4)

Tb4b4b4}

Te6edf3awaitClose Tb4b4b4{ Te6edf3disconnectTb4b4b4(Tb4b4b4) Tb4b4b4}

Tb4b4b4}

Tff7b72private Tff7b72fun Td2a8ffsubscribeTb4b4b4(Te6edf3topicTb4b4b4: Tffa657StringTb4b4b4) Tb4b4b4{

Te6edf3mqttClientTff7b72?.Te6edf3subscribeTb4b4b4(Te6edf3topicTb4b4b4, Te6edf3DEFAULT_QOSTb4b4b4)

Te6edf3TimberTb4b4b4.Te6edf3iTb4b4b4(Ta5d6ff"Ta5d6ffMQTT Subscribed to topic: Tffd700$Te6edf3topicTa5d6ff"Tb4b4b4)

Tb4b4b4}

Tff7b72fun Td2a8ffpublishTb4b4b4(Te6edf3topicTb4b4b4: Tffa657StringTb4b4b4, Te6edf3dataTb4b4b4: Te6edf3ByteArrayTb4b4b4, Te6edf3retainedTb4b4b4: Tffa657BooleanTb4b4b4) Tb4b4b4{

Tff7b72try Tb4b4b4{

Tff7b72val Te6edf3token Tff7b72= Te6edf3mqttClientTff7b72?.Te6edf3publishTb4b4b4(Te6edf3topicTb4b4b4, Te6edf3dataTb4b4b4, Te6edf3DEFAULT_QOSTb4b4b4, Te6edf3retainedTb4b4b4)

Te6edf3TimberTb4b4b4.Te6edf3iTb4b4b4(Ta5d6ff"Ta5d6ffMQTT Publish messageId: Tffd700${Te6edf3tokenTff7b72?.Te6edf3messageIdTffd700}Ta5d6ff"Tb4b4b4)

Tb4b4b4} Tff7b72catch Tb4b4b4(Te6edf3exTb4b4b4: Te6edf3ExceptionTb4b4b4) Tb4b4b4{

Te6edf3TimberTb4b4b4.Te6edf3eTb4b4b4(Ta5d6ff"Ta5d6ffMQTT Publish error: Tffd700${Te6edf3exTb4b4b4.Te6edf3messageTffd700}Ta5d6ff"Tb4b4b4)

Tb4b4b4}

Tb4b4b4}

Tb4b4b4}